pytest-xdistでテストを並列実行するとき、sessionのfixtureを1度だけ実行させる
2024.03.27pytestのプラグインであるpytest-xdistを利用すると、テストを並列実行できます。これにより、テスト時間の短縮が見込めます。
しかし、並列実行するため、sessionスコープ(1度だけ実行される)のfixtureも並列数だけ実行されます。 そのため、次のような場合で課題となる可能性もあります。
- テスト用データの用意などで、時間がかかる
- 外部APIを利用するため、rate limitの上限が気になる
- など
たとえば、APIの認可にAuth0を利用している場合、E2Eテスト用のアクセストークンの取得（M2M認証）は、月1000回の制限があります（プランにもよる）。
テストの実行回数や並列数にもよりますが、少し心配になります。（1日あたり50回＆10並列としたとき、1日あたり5回まで実行できる）
そこで、並列実行時でもsessionスコープのfixtureを1度だけ実行する方法を試してみました。
おすすめの方
- pytestで並列実行したい方
- pytestの並列実行でsessionスコープのfixtureを1度だけ実行したい方
本記事の補足
公式ドキュメントの紹介を参考にしています。
1度だけ実行するsessionスコープのfixtureは、Auth0のアクセストークン取得を想定します。 そのため、各プロセスで同じ値を参照させたいですが、環境変数にアクセストークンを保存するのはセキュリティ的に怖いため、ローカルのtempフォルダに保存します。 セキュリティをさらに考慮する場合は、一時保存場所をAWS Systems Managerのパラメータストアなどにすると良さそうです。
実験環境を構築する
python -m venv .venv source .venv/bin/activate pip install pytest pytest-xdist filelock
並列実行したとき、sessionのfixtureが複数実行することを確認する（アクセストークン取得）
pytest-xdistを利用すると通常のprintが表示されないため、実験用のログ確認のためstderrに出力しています。
conftest.py
1度だけ実行したいsessionの関数として、「token()」を用意しました。 Auth0からE2Eテスト用のアクセストークンを取得する想定です。
また、sessionとfunctionのスコープを毎回実行させています。
import random
import string
import sys
import pytest
@pytest.fixture(scope="session")
def token():
# tokenを取得したことにする
return "".join(random.choices(string.ascii_letters + string.digits, k=5))
@pytest.fixture(scope="session", autouse=True)
def my_setup_session():
print("my_setup_session", file=sys.stderr)
@pytest.fixture(scope="function", autouse=True)
def my_setup_function():
print("my_setup_function", file=sys.stderr)
test_e2e.py
E2Eテスト用のコードです。受け取ったtokenを出力するだけです。 本来はこの部分で、任意のAPIに対してリクエストする想定です。
import sys
import pytest
def test_1(token):
print(token, file=sys.stderr)
assert True
def test_2(token):
print(token, file=sys.stderr)
assert True
def test_3(token):
print(token, file=sys.stderr)
assert True
def test_4(token):
print(token, file=sys.stderr)
assert True
def test_5(token):
print(token, file=sys.stderr)
assert True
テストを並列実行する
worker数は3つにしてみました。
pytest -v --capture=no -n 3
結果は下記です。改行的な問題で見づらいですが、sessionの関数が3回実行されており、tokenの値も3種類あります。
plugins: xdist-3.4.0 3 workers [5 items] scheduling tests via LoadScheduling my_setup_session test_e2e.py::test_3 my_setup_session my_setup_session my_setup_function my_setup_function my_setup_function test_e2e.py::test_2 test_e2e.py::test_1 mmVGf mvwyD KNe0n my_setup_function my_setup_function KNe0n mvwyD [gw2] PASSED test_e2e.py::test_3 [gw1] PASSED test_e2e.py::test_2 [gw0] PASSED test_e2e.py::test_1 test_e2e.py::test_4 [gw0] PASSED test_e2e.py::test_4 test_e2e.py::test_5 [gw1] PASSED test_e2e.py::test_5
sessionスコープのfixtureを1度だけ実行させる
conftest.py
公式ドキュメントを参考にして、token()を更新します。
「tmp_path_factory」は、pytestが用意している一時作業用のディレクトリを作成するfixtureです。
import random
import string
import sys
import pytest
from filelock import FileLock
from time import sleep
@pytest.fixture(scope="session")
def token(tmp_path_factory, worker_id):
def get_token():
# tokenを取得したことにする（取得まで2秒かかる想定とする）
sleep(2)
return "".join(random.choices(string.ascii_letters + string.digits, k=5))
if worker_id == "master":
# 並列実行でない場合は、tokenを取得して返す
return get_token()
root_tmp_dir = tmp_path_factory.getbasetemp().parent
fn = root_tmp_dir / "token.txt"
with FileLock(str(fn) + ".lock"):
if fn.is_file():
return fn.read_text()
token = get_token()
fn.write_text(token)
return token
@pytest.fixture(scope="session", autouse=True)
def my_setup_session():
print("my_setup_session", file=sys.stderr)
@pytest.fixture(scope="function", autouse=True)
def my_setup_function():
print("my_setup_function", file=sys.stderr)
テストを並列実行すると、同じtokenの値になっている
並列実行したすべてのテストで「同じtoken」を取得できました。 動作的には、token取得時にログの表示が約2秒ほど停止していました（get_token()が終了するまで待ってる）。
plugins: xdist-3.4.0 3 workers [5 items] scheduling tests via LoadScheduling my_setup_session test_e2e.py::test_3 test_e2e.py::test_2 my_setup_session test_e2e.py::test_1 my_setup_session my_setup_function fYXrk [gw2] PASSED test_e2e.py::test_3 my_setup_function fYXrk [gw0] PASSED test_e2e.py::test_1 test_e2e.py::test_4 my_setup_function fYXrk [gw0] PASSED test_e2e.py::test_4 my_setup_function fYXrk [gw1] PASSED test_e2e.py::test_2 test_e2e.py::test_5 my_setup_function fYXrk [gw1] PASSED test_e2e.py::test_5
一時フォルダを確認すると次のようになっていました。
- テスト実行のたびに、新しいフォルダが作成される
- テスト実行のたびに、古いフォルダが削除される
